package com.atguigu.flink.exec2;

import com.atguigu.flink.pojo.AdsClickLog;
import com.atguigu.flink.pojo.WaterSensor;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.RuntimeContext;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

import java.text.SimpleDateFormat;
import java.util.Date;

/**
 * Created by Smexy on 2022/11/25
 *
 *      需求： 1天内，统计如果用户点击广告的次数。
 *              如果超过100，拉黑，后续的点击不统计了。
 *
 *              1天后，重置！
 *
 *      思路:
 *              前提： 点击行为是按照时间顺序排好的。需要感知日期的变化。
 *                  a,点击 1号广告， 6-11
 *                  a,点击 1号广告， 6-12   认为日期已经变化
 *
 *
 *              存储上一条数据的日期，用于和当前数据的日期比较，感知时间的变化，日期变化后，拉黑技术清零。
 *              存储每个用户点击每个广告在当天的次数
 *              存储每个用户是否已经被拉黑，拉黑，后续的点击不统计了。

 */
public class Demo4_ClickAds
{
    public static void main(String[] args) {


        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(1);

        env
           .readTextFile("data/AdClickLog.csv")
           .map(new MapFunction<String, AdsClickLog>()
           {
               @Override
               public AdsClickLog map(String value) throws Exception {
                   String[] words = value.split(",");
                   return new AdsClickLog(
                       Long.valueOf(words[0]),
                       Long.valueOf(words[1]),
                       words[2],
                       words[3],
                       Long.valueOf(words[4])
                   );
               }
           })
           // 按照用户 -- 广告
           .keyBy(log -> log.getUserId()+"_"+log.getAdsId())
           .process(new KeyedProcessFunction<String, AdsClickLog, String>()
           {


               @Override
               public void open(Configuration parameters) throws Exception {

                   RuntimeContext context = getRuntimeContext();
                   ifBlack = context.getState(new ValueStateDescriptor<Boolean>("ifBlack", Boolean.class));
                   count = context.getState(new ValueStateDescriptor<Integer>("count", Integer.class));
                   lastDate = context.getState(new ValueStateDescriptor<String>("lastDate", String.class));

               }

               //使用键控状态
               private ValueState<String> lastDate;
               private ValueState<Integer> count;
               private ValueState<Boolean> ifBlack;

                //是否需要做成状态
               // 如果做成状态，一定是一个 键控状态，每一个key，都有一个format，没有必要
               // 不用做成状态(非只读属性)  format是只读的
               SimpleDateFormat format = new SimpleDateFormat("yyyy-MM-dd");
               @Override
               public void processElement(AdsClickLog value, Context ctx, Collector<String> out) throws Exception {

                   //判断当前日期是否要重置计数
                   String lastDateStr = lastDate.value();

                   String currentDate = this.format.format(new Date(value.getTimestamp() * 1000));

                    //当第一条数据达到，及日期更新，做初始化，重置
                   if (lastDateStr == null || !currentDate.equals(lastDateStr)){
                       //日期推进，要重置计数
                       lastDate.update(currentDate);
                       count.update(0);
                       ifBlack.update(false);
                   }

                   //定义输出的格式
                   String msg = value.getUserId() + "点击了 " + value.getAdsId()  +",";

                   //先更新点击次数
                   count.update( count.value() + 1);
                   //判断是否已经拉黑
                   if (!ifBlack.value() && count.value() >= 100 ){

                       //在第一次拉黑时输出消息，此后再点击，不计数，也不再输出
                       out.collect(msg + count.value() + ",被拉黑");

                       ifBlack.update(true);

                   }else{
                       if (count.value() < 100 ){
                           //没有拉黑
                           out.collect(msg + count.value());
                       }

                   }

               }
           }).print().setParallelism(1);

        try {
                    env.execute();
                } catch (Exception e) {
                    e.printStackTrace();
                }

    }
}
